%%%====================================================================
%%% File    : pipe.erl
%%% Author  :  <vjache>
%%% Description: TODO: Add description to pipe
%%%
%%% Created : 23.03.2009
%%%====================================================================

-module(pipe).

%%
%% Include files
%%
-include_lib("eunit/include/eunit.hrl").
-include("../include/pipe.hrl").

%%
%% Exported Functions
%%
-export([
		 open/1,
		 push/2,
		 flush/1,
		 close/1,
		 push_each_prog/3,
		 push_each/2,
		 push_each/3,
		 pipe_from_module/3,
		 terminator/0,
		 trivial_consumer/1,
		 trivial_consumer/2,
		 trivial_consumer/3,
		 simple_immutable/3,
		 from_fun/2]).

%%====================================================================
%% API Functions
%%====================================================================

%%--------------------------------------------------------------------
%% Function: 	open(Pipe) -> PipeState
%% Description: Opens a pipe by descriptor. Accepts an pipe
%%				description record and returns an initial state of
%%				pipe.
%%--------------------------------------------------------------------
open(Pipe) when is_record(Pipe,pipe)-> %%PipeState
    OpenFun=Pipe#pipe.open_fun,
	{Pipe,OpenFun({Pipe#pipe.param,Pipe#pipe.next_pipe})}.
%%--------------------------------------------------------------------
%% Function: 	push(PipeState,Item) -> NewPipeState
%% Description: Pushes a next item to the pipe and moves the pipe 
%%				state further. Accepts a current pipe state and 
%%				returns a new pipe state.
%%--------------------------------------------------------------------
push({Pipe,State},Item) ->
    PushFun=Pipe#pipe.push_fun,
	{Pipe,PushFun(State,Item)}.
%%--------------------------------------------------------------------
%% Function: 	flush(PipeState) -> NewPipeState
%% Description: Flushes pipe to the uderlying pipe or io device.
%%--------------------------------------------------------------------
flush({Pipe,State}) ->
    FlushFun=Pipe#pipe.flush_fun,
	{Pipe,FlushFun(State)}.
%%--------------------------------------------------------------------
%% Function: 	close(PipeState) -> ok
%% Description: Closes a pipe. After this call the pipe state 
%%				could not be used in such a calls like push, flush, etc.
%%--------------------------------------------------------------------
close({Pipe,_State}=S) when is_record(Pipe,pipe) ->
%% 	{Pipe,State}=flush(push(S,?EOF)),
	{Pipe,State}=flush(S),
	CloseFun=Pipe#pipe.close_fun,
	CloseFun(State),
	ok.

push_each_prog(Pipe,IterOrState,ProgFun) when is_record(Pipe, pipe)->
	push_each(pipe:open(Pipe),IterOrState,ProgFun).
push_each(Pipe,IterOrState) when is_record(Pipe, pipe)->
	push_each(pipe:open(Pipe),IterOrState);
push_each(PipeState,IterOrState) ->
	push_each(PipeState,IterOrState,fun(_)->ok end).
push_each(PipeState,IterOrState,ProgFun) ->
	{IterStateFin,{PipeStateFinal,_}}=iter:fold(IterOrState, 
			  fun(Item,{AccIn,Count})->
					  ProgFun(Count),
					  {pipe:push(AccIn,Item),Count+1}
			  end, {PipeState,0}),
	{IterStateFin,PipeStateFinal}.

pipe_from_module(Mod,Param,NextPipe) ->
	#pipe{
		  param=Param,
		  open_fun=fun(Pipe) -> Mod:open(Pipe) end,
		  push_fun=fun(PipeState,Item) -> Mod:push(PipeState,Item) end,
		  flush_fun=fun(PipeState) -> Mod:flush(PipeState) end,
		  close_fun=fun(PipeState) -> Mod:close(PipeState) end,
		  next_pipe=NextPipe}.

terminator() ->
	#pipe{
		  param=terminator,
		  open_fun=fun(_) -> terminator end,
		  push_fun=fun(terminator,_Item) -> terminator end,
		  flush_fun=fun(terminator) -> terminator end,
		  close_fun=fun(terminator) -> terminator end,
		  next_pipe=terminal}.
trivial_consumer(ConsumeFun) when is_function(ConsumeFun,1) ->
	#pipe{
		  param=trivial_consumer,
		  open_fun=fun(_) -> trivial_consumer end,
		  push_fun=fun(trivial_consumer,Item) -> ConsumeFun(Item),trivial_consumer end,
		  flush_fun=fun(trivial_consumer) -> trivial_consumer end,
		  close_fun=fun(trivial_consumer) -> trivial_consumer end,
		  next_pipe=trivial_consumer}.
trivial_consumer(ConsumeFun, CloseFun) 
  when is_function(ConsumeFun,1),is_function(CloseFun,0) ->
	#pipe{
		  param=trivial_consumer,
		  open_fun=fun(_) -> trivial_consumer end,
		  push_fun=fun(trivial_consumer,Item) -> ConsumeFun(Item),trivial_consumer end,
		  flush_fun=fun(trivial_consumer) -> trivial_consumer end,
		  close_fun=fun(trivial_consumer) -> CloseFun(),ok end,
		  next_pipe=trivial_consumer}.
trivial_consumer(OpenFun,ConsumeFun, CloseFun) 
  when is_function(ConsumeFun,1),is_function(OpenFun,0),is_function(CloseFun,0) ->
	#pipe{
		  param=trivial_consumer,
		  open_fun=fun(_) -> OpenFun(),trivial_consumer end,
		  push_fun=fun(trivial_consumer,Item) -> ConsumeFun(Item),trivial_consumer end,
		  flush_fun=fun(trivial_consumer) -> trivial_consumer end,
		  close_fun=fun(trivial_consumer) -> CloseFun(),ok end,
		  next_pipe=trivial_consumer}.
simple_immutable(OpenFun,ConsumeFun, CloseFun) 
  when is_function(ConsumeFun,2),is_function(OpenFun,0),is_function(CloseFun,1) ->
	#pipe{
		  param=simple_immutable,
		  open_fun=fun(_) -> OpenFun() end,
		  push_fun=fun(S,Item) -> ConsumeFun(S,Item),S end,
		  flush_fun=fun(S) -> S end,
		  close_fun=fun(S) -> CloseFun(S),ok end,
		  next_pipe=simple_immutable}.

from_fun(State0,TransitionFun) when 
  is_function(TransitionFun,2) ->
	#pipe{
		  param=nil,
		  open_fun=fun(_) -> TransitionFun(State0,open) end,
		  push_fun=fun(S,Item) -> TransitionFun(S,Item) end,
		  flush_fun=fun(S) -> TransitionFun(S,flush) end,
		  close_fun=fun(S) -> TransitionFun(S,close) end,
		  next_pipe=nil}.
	

%%====================================================================
%% Local Functions
%%====================================================================

from_fun_test() ->
	Pipe=pipe:from_fun(
		   {1,pipe_console:new()}, 
		   fun
			  ({1,Pipe},open) ->
				   {1,pipe:open(Pipe)};
			  ({_S,PS},flush) ->
				   {_S,pipe:flush(PS)};
			  ({_S,PS},close) ->
				   {_S,pipe:close(PS)};
			  ({S,PS},I) ->
				   {S+I,pipe:push(PS,S+I)}
		   end),
	Pump=pump:new(iter:from_list([1,2,3,4,5,6]), Pipe),
	pump:init(Pump).

